package org.pizza.rocket.core.consumer;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import org.pizza.rocket.core.annotation.RocketMessageListener;
import org.pizza.rocket.core.listener.SimpleRocketMQListener;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.*;

/**
 * @author 高巍
 * email kynb_club@163.com
 * @since 2020/11/23 3:36 下午
 * 普通消费者包装类
 */
@Slf4j
public class SimpleConsumerWrapper extends AbstractConsumerWrapper {
    /**
     * 原生的消费者bean
     */
    private ConsumerBean consumer;
    /**
     * 自己封装直接接收泛型的监听器
     */
    private SimpleRocketMQListener mqListener;

    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();
        this.messageType = getMessageType(this.mqListener);
        if (log.isInfoEnabled()) {
            log.info("RocketMQ messageType: {}", messageType.getName());
        }
    }

    private void initConsumer() {
        this.consumer = new ConsumerBean();
        //配置
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.AccessKey, environment.resolvePlaceholders(annotation.accessKey()));
        properties.setProperty(PropertyKeyConst.SecretKey, environment.resolvePlaceholders(annotation.secretKey()));
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, environment.resolvePlaceholders(annotation.nameServer()));
        properties.setProperty(PropertyKeyConst.GROUP_ID, environment.resolvePlaceholders(annotation.groupName()));
        properties.setProperty(PropertyKeyConst.MessageModel, annotation.messageModel());
        //将消费者线程数（20为默认值）
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, String.valueOf(annotation.consumeThreadNums()));
        consumer.setProperties(properties);
        //订阅关系表
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        Subscription subscription = new Subscription();
        subscription.setTopic(environment.resolvePlaceholders(annotation.topic()));
        subscription.setExpression(environment.resolvePlaceholders(annotation.tag()));
        subscriptionTable.put(subscription, new DefaultMessageListener());
        //订阅多个topic如上面设置
        consumer.setSubscriptionTable(subscriptionTable);
    }

    @Override
    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("Consumer already running. " + this.toString());
        }
        this.consumer.start();
        this.setRunning(true);
    }

    @Override
    public void stop(Runnable callback) {
        stop();
        callback.run();
    }

    @Override
    public void stop() {
        if (this.isRunning()) {
            if (Objects.nonNull(consumer)) {
                consumer.shutdown();
            }
            setRunning(false);
        }
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public boolean isRunning() {
        return running;
    }


    @Override
    public void destroy() throws Exception {
        this.setRunning(false);
        if (Objects.nonNull(consumer)) {
            consumer.shutdown();
        }
        log.info("customer destroyed, {}", this.toString());
    }

    @Override
    public int getPhase() {
        // Returning Integer.MAX_VALUE only suggests that
        // we will be the first bean to shutdown and last bean to start
        return Integer.MAX_VALUE;
    }

    /**
     * 默认的消息监听器
     */
    public class DefaultMessageListener implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext context) {
            Object msg = null;
            try {
                MDC.put("X-B3-TraceId", UUID.randomUUID().toString().split("-")[0]);
                msg = doConvertMessage(message);
                if (message.getReconsumeTimes() > 0) {
                    log.info("received message ,current reconsume times :{} msg: {} ", message.getReconsumeTimes(), msg);
                }
                long now = System.currentTimeMillis();
                mqListener.onMessage(msg);
                long costTime = System.currentTimeMillis() - now;
                if (log.isDebugEnabled()) {
                    log.debug("consume {} cost: {} ms", message.getMsgID(), costTime);
                }
                return Action.CommitMessage;
            } catch (Exception e) {
                log.warn("consume message failed，current reconsume times :{} msg: {} ", message.getReconsumeTimes(), msg, e);
                if (message.getReconsumeTimes() > annotation.maxReconsumeTimes()) {
                    mqListener.onMaxRetry(message,msg);
                    return Action.CommitMessage;
                }
                mqListener.onError(message, msg, e);
                return Action.ReconsumeLater;
            } finally {
                MDC.clear();
            }
        }
    }

    public SimpleRocketMQListener getMqListener() {
        return mqListener;
    }

    public void setMqListener(SimpleRocketMQListener mqListener) {
        this.mqListener = mqListener;
    }

    public RocketMessageListener getAnnotation() {
        return annotation;
    }

    public void setAnnotation(RocketMessageListener annotation) {
        this.annotation = annotation;
    }

}
